#include "StdAfx.h"
#include "CommandManager.h"
#include "IOManager.h"
#include "Types.h"
#include "json.h"
#include "RandomGeneratedStreamInput.h"
#include "JsonStreamServer.h"
#include "SocketStreamInput.h"
#include "QueryIntermediateRepresentation.h"
#include "SocketStreamOutput.h"
#include "IStreamOutput.h"
#include "IStreamInput.h"
#include "ShowResultStreamOutput.h"
CommandManager* CommandManager::commandManager = NULL;
CommandManager::CommandManager(void)
{
	initial();
}


CommandManager::~CommandManager(void)
{
}
CommandManager* CommandManager::getInstance()
{
	if(commandManager==NULL)
	{
		commandManager = new CommandManager();
	}
	return commandManager;
}
void CommandManager::initial()
{
	IOManager::getInstance()->addCommandInput(SERVER_IP,SERVER_PORT,this);
}
//some kinds of command:
//1.register schema
//2.register
bool CommandManager::processCommand(char* command,size_t length)
{
	std::string ret = "";
	int currentSize = 0;
	
	while(currentSize < length)
	{
		std::string cmd(command + currentSize);
		//std::cout<<cmd<<std::endl;
	    Document commandDocument = fromjson(cmd);
		
		std::string commandType = commandDocument.getField(COMMAND_TYPE).valuestr();
		if(commandType == REGISTER_SCHEMA)
		{
			processRegisterSchemaCommand(commandDocument);
		}
		else if(commandType == REGISTER_QUERY)
		{
			processRegisterQueryCommand(commandDocument);
		}
		else
		{
			assert(false); // never reached
			return false;
		}
	
		currentSize += cmd.length() + 1;
		
	}
	return true;
	 
}
bool CommandManager::processRegisterSchemaCommand(Document& document)
{
	Document& schemaDocument = document.getField(SCHEMA_CONTENT).embeddedObject();
	boost::shared_ptr<JsonSchema> jsonSchema(new JsonSchema(schemaDocument));

	Document& inputSpecificationDocument = document.getField(INPUT_SPECIFICATION).embeddedObject();
	std::string inputType =  inputSpecificationDocument.getField(INPUT_TYPE).valuestr();
	if(inputType == RANDOM_GENERATED)
	{
		boost::shared_ptr<IStreamInput> streamInput (new RandomGeneratedStreamInput(jsonSchema));
		JsonStreamServer::getInstance()->registerStream(streamInput);
	}
	else if(inputType == SOCKET_INPUT)
	{
		std::string ip = inputSpecificationDocument.getField(INPUT_IP).valuestr();
		std::string port = inputSpecificationDocument.getField(INPUT_PORT).valuestr();
		boost::shared_ptr<IStreamInput> streamInput(new SocketStreamInput(ip,port,jsonSchema));
		JsonStreamServer::getInstance()->registerStream(streamInput);

	}
	else 
	{
		assert(false);//never reached
		return false;
	}
	 
	
	return true;
}
bool CommandManager::processRegisterQueryCommand(Document& document)
{
	Document& queryDocument = document.getField(QUERY_CONTENT).embeddedObject();
	boost::shared_ptr<QueryIntermediateRepresentation>intermediateQuery(new QueryIntermediateRepresentation(queryDocument));
	Document& outputSpecificationDocument = document.getField(OUTPUT_SPECIFICATION).embeddedObject();
	std::string outputType =  outputSpecificationDocument.getField(OUTPUT_TYPE).valuestr();
	if(outputType == SHOW_RESULT)
	{
		boost::shared_ptr<IStreamOutput> streamOutput(new ShowResultStreamOutput());
		JsonStreamServer::getInstance()->registerQuery(intermediateQuery,streamOutput);
	}
	else if(outputType == SOCKET_OUTPUT)
	{
		std::string ip = outputSpecificationDocument.getField(OUTPUT_IP).valuestr();
		std::string port = outputSpecificationDocument.getField(OUTPUT_PORT).valuestr();
		boost::shared_ptr<IStreamOutput> streamOutput(new SocketStreamOutput(ip,port));
		JsonStreamServer::getInstance()->registerQuery(intermediateQuery,streamOutput);

	}
	else 
	{
		assert(false);//never reached
		return false;
	}
	
	
	
	return true;
}